package design_patterns_and_best_practices_in_java;

import io.reactivex.Observable;
import io.reactivex.observables.GroupedObservable;
import io.reactivex.schedulers.TestScheduler;

import java.io.IOException;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @author lzy
 * @version v1.0
 * Create DateTime: 2019/11/17 15:41
 */
public class Chapter6RxJava {
    public static void main(String[] args) throws IOException {
        Observable.just("hello world");

        Observable.create(obersver -> {
            try {
                if (!obersver.isDisposed()) {
                    for (int i = 0; i < 5; i++) {
                        obersver.onNext(i);
                    }
                    obersver.onComplete();
                }
            } catch (Exception e) {
                obersver.onError(e);
            }
        }).subscribe(System.out::println, System.out::println, () -> {
            System.out.println("Sequence complete.");
        });

        final Observable<Integer> defer = Observable.defer(() -> Observable.just(123));
        defer.subscribe(System.out::println);

        final Observable<String> stringObservable = Observable.fromArray("a", "b", "c");
        stringObservable.subscribe(System.out::println);

        /*Observable.interval(1, TimeUnit.SECONDS).map(tick -> tick.longValue()).subscribe(System.out::println
        );*/

        final Integer[] x = {1};
        Observable.just("a").repeatUntil(() -> x[0]++ > 2).subscribe(System.out::println);

        Observable.range(0, 10).buffer(6).subscribe(System.out::println);

        TestScheduler testScheduler = new TestScheduler();
        Observable<String> abc = Observable.fromArray("a", "b", "c");
        abc.flatMap(y -> Observable.just(y + "f"))
                .delay(new Random().nextInt(5), TimeUnit.SECONDS, testScheduler)
                .toList().subscribe(System.out::println, System.out::println);
        testScheduler.advanceTimeBy(30, TimeUnit.SECONDS);

        Observable<String> list = Observable.fromArray("aaa", "baa", "ac", "ccc", "ccs");
        list.groupBy(y -> y.substring(0, 1)).subscribe(z ->
        {
            GroupedObservable<String, String> g = z;
            System.out.println("----" + g.getKey() + "----");
            g.subscribe(System.out::println);
        });

        Observable.range(1, 5).scan((z, sum) -> z + sum).subscribe(System.out::println);

        Observable.range(1, 5).window(1).flatMap(z -> z.scan((y, s) -> y + s)).subscribe(System.out::println);

        list.filter(y -> y.startsWith("a")).subscribe(System.out::println);

        final Observable<Long> a = Observable.interval(6, TimeUnit.SECONDS);
        final Observable<Long> b = Observable.interval(10, TimeUnit.SECONDS);
        Observable.combineLatest(a, b, (z, y) -> z.toString() + "-" + y.toString()).blockingForEach(System.out::println);

        System.in.read();
    }
}
